package cn.dmp.tags

import ch.hsr.geohash.GeoHash
import org.apache.commons.lang.StringUtils
import org.apache.spark.sql.Row
import redis.clients.jedis.Jedis

object Tags4Business extends Tags{
    /**
      * 行数据 -》 标签数据
      *
      * @param args
      * @return
      */
    override def makeTags(args: Any*): Map[String, Int] = {

        var tags = Map[String, Int]()

        val row = args(0).asInstanceOf[Row]
        val jedis = args(1).asInstanceOf[Jedis]

        import cn.dmp.beans.RichString._

        val lat = row.getAs[String]("lat").toDoublePlus
        val long = row.getAs[String]("long").toDoublePlus

        if (lat>=3.86 && lat <= 53.55 && long>=73.66 && long <= 135.05) {
            val geoHashCode = GeoHash.withCharacterPrecision(lat, long, 8).toBase32
            val bTag = jedis.hget("busdict", geoHashCode)
            if (StringUtils.isNotEmpty(bTag)) tags += "BS"+bTag -> 1
        }
        tags
    }
}
